package com.xctech.iot.rocketmq.producer;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.UUID;

/**
 * Rocketmq生产者
 *
 * @author: feihu.liang
 * @since: 2024/4/11 11:09
 */
@Slf4j
@Service
public class RocketMQProducer {

	@Autowired
	private RocketMQTemplate rocketMQTemplate;

	/**
	 * 发送普通消息
	 *
	 * @param topic   topic
	 * @param message 消息体
	 */
	public void sendMessage(String topic, Object message) {
		this.rocketMQTemplate.convertAndSend(topic, message);
	}

	/**
	 * 发送同步消息
	 *
	 * @param topic   topic
	 * @param message 消息体
	 */
	public void syncSendMessage(String topic, Object message) {
		this.rocketMQTemplate.syncSend(topic, message);
	}

	/**
	 * 发送异步消息
	 *
	 * @param topic   topic
	 * @param message 消息体
	 */
	public void asyncSendMessage(String topic, Object message) {
		this.rocketMQTemplate.asyncSend(topic, message, new SendCallback() {
			@Override
			public void onSuccess(SendResult sendResult) {
				log.info("异步消息发送成功，message = {}, SendStatus = {}", message, sendResult.getSendStatus());
			}

			@Override
			public void onException(Throwable e) {
				log.error("异步消息发送异常 topic={} exception=", topic, e);
			}
		});
	}


	/**
	 * 同步发送批量消息
	 *
	 * @param topic       topic
	 * @param messageList 消息集合
	 * @param timeout     超时时间（毫秒）
	 */
	public void syncSendMessages(String topic, List<Message<?>> messageList, long timeout) {
		this.rocketMQTemplate.syncSend(topic, messageList, timeout);
		log.info("同步发送批量消息完成：topic={} message={}", topic, JSON.toJSONString(messageList));
	}

	/**
	 * 发送事务消息
	 *
	 * @param topic   topic
	 * @param message 消息对象
	 */
	public void sendMessageInTransaction(String topic, Object message) {
		String transactionId = UUID.randomUUID().toString();
		TransactionSendResult result = this.rocketMQTemplate.sendMessageInTransaction(topic, MessageBuilder.withPayload(message)
				.setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId)
				.build(), message);
		log.info("发送事务消息（半消息）完成：result = {}", result);
	}

	/**
	 * 同步发送延时消息
	 *
	 * @param topic      topic
	 * @param message    消息体
	 * @param timeout    超时
	 * @param delayLevel 延时等级：现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，
	 *                   从1s到2h分别对应着等级 1 到 18，消息消费失败会进入延时消息队列
	 *                   "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
	 */
	public void syncSendDelay(String topic, Object message, long timeout, int delayLevel) {
		this.rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(message).build(), timeout, delayLevel);
		log.info("已同步发送延时消息 message = {}", message);
	}

	/**
	 * 异步发送延时消息
	 *
	 * @param topic      topic
	 * @param message    消息对象
	 * @param timeout    超时时间
	 * @param delayLevel 延时等级
	 */
	public void asyncSendDelay(String topic, Object message, long timeout, int delayLevel) {
		this.rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(message).build(), new SendCallback() {
			@Override
			public void onSuccess(SendResult sendResult) {
				log.info("异步发送延时消息成功，message = {}", message);
			}

			@Override
			public void onException(Throwable throwable) {
				log.error("异步发送延时消息发生异常，exception = {}", throwable.getMessage());
			}
		}, timeout, delayLevel);
		log.info("已异步发送延时消息 message = {}", message);
	}

}
